package com.suxin.test;

import com.suxin.mqtt.callback.PushCallback;
import com.suxin.mqtt.config.MqttInputConfig;
import com.suxin.mqtt.config.MyGateway;
import com.suxin.mqtt.domain.MqMessage;
import com.suxin.mqtt.util.AbstractConsumer;
import com.suxin.mqtt.util.MqttClientProxy;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.junit.jupiter.api.Test;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @Classname App
 * @Description [ TODO ]
 * @Author Tang
 * @Date 2020/10/4 17:28
 * @Created by ASUS
 */
//@SpringBootTest(classes = SpringMqttApplication.class)
public class App {

    @Resource
    MyGateway myGateway;

    @Resource
    MqttInputConfig mqttInputConfig;

    @Test
    public void send() {
        mqttInputConfig.getAdapter().addTopic("test", 1);
        myGateway.sendToMqtt("hello world!!", "mytopic");
        myGateway.sendToMqtt("hello world!!", "test");
    }

    public static void main(String[] args) {

        new App().send();
    }

    @Test
    public void test03() {
        MqttClientProxy mqttClientProxy = new MqttClientProxy("tcp://192.168.108.128:1883",
                "", "");
        mqttClientProxy.register("repub/to/1", new AbstractConsumer((mqAggregation) -> {
            System.out.println("接收到的消息为：" + mqAggregation.getPayload());
        }));

        // $SYS/brokers/+/clients/#

//        mqttClientProxy.register("$SYS/#", new AbstractConsumer((mqAggregation) -> {
//            System.out.println("接收到的消息为：" + mqAggregation.getPayload());
//        }));
        mqttClientProxy.publish(new MqMessage("$delayed/10/test/1", "hello testtopic/1! 10~~"));
        try {
            System.out.println("等待延迟接收...");
            TimeUnit.SECONDS.sleep(15);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        mqttClientProxy.disconnect();
    }


    static class OnMessageCallback implements MqttCallback {

        @Override
        public void connectionLost(Throwable throwable) {
            System.out.println("连接断开，可以做重连");
        }

        @Override
        public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
            System.out.println("接收到的消息主题：" + topic);
            System.out.println("接收消息Qos: " + mqttMessage.getQos());
            System.out.println("接收消息内容：" + new String(mqttMessage.getPayload()));
        }

        @Override
        public void deliveryComplete(IMqttDeliveryToken token) {
            System.out.println("deliveryComplete---------" + token.isComplete());
        }
    }

    @Test
    public void test04() throws Exception {

        String subTopic = "test/#";
        String pubTopic = "test/1";

        String content = "hello world! hello!";

        int qos = 2;
        MemoryPersistence persistence = new MemoryPersistence();

        MqttClient mqttClient = new MqttClient("tcp://192.168.108.128:1883", "emqt_test", persistence);

        MqttConnectOptions options = new MqttConnectOptions();
        // 保留会话
        options.setCleanSession(true);
        // 设置回调
        mqttClient.setCallback(new OnMessageCallback());
        // 建立连接
        mqttClient.connect(options);
        // 订阅主题
        mqttClient.subscribe(subTopic);

        MqttMessage mqttMessage = new MqttMessage(content.getBytes());
        mqttMessage.setQos(2);

        mqttClient.publish(pubTopic, mqttMessage);
        System.out.println("发布消息。。。。");

        mqttClient.disconnect();
        mqttClient.close();

    }

    @Test
    public void test02() {
        test01();
    }

    private static void test01() {
        String subTopic = "testtopic/#";
        String pubTopic = "testtopic/1";
        String content = "Hello World";

        int qos = 2;
        String broker = "tcp://192.168.108.128:1883";
        String clientId = "emqx_test";
        MemoryPersistence persistence = new MemoryPersistence();

        try {
            MqttClient client = new MqttClient(broker, clientId, persistence);

            // MQTT 连接选项
            MqttConnectOptions connOpts = new MqttConnectOptions();
//            connOpts.setUserName("emqx_test");
//            connOpts.setPassword("emqx_test_password".toCharArray());
            // 保留会话
            connOpts.setCleanSession(true);

            // 设置回调
            client.setCallback(new PushCallback());

            // 建立连接
            System.out.println("Connecting to broker: " + broker);
            client.connect(connOpts);

            System.out.println("Connected");
            System.out.println("Publishing message: " + content);

            // 订阅
            client.subscribe(subTopic);

            // 消息发布所需参数
//            MqttMessage message = new MqttMessage(content.getBytes());
//            message.setQos(qos);
//            client.publish(pubTopic, message);
//            System.out.println("Message published");

            client.disconnect();
            System.out.println("Disconnected");
            client.close();
            System.exit(0);
        } catch (MqttException me) {
            System.out.println("reason " + me.getReasonCode());
            System.out.println("msg " + me.getMessage());
            System.out.println("loc " + me.getLocalizedMessage());
            System.out.println("cause " + me.getCause());
            System.out.println("excep " + me);
            me.printStackTrace();
        }
    }

}